package paas.storage.distributedColumnDatabase.impl;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.io.Writable;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;


/**
 * @author xufeng
 */
public class orc {
    String path;

    /**
     * 导入orc文件
     *
     * @param path
     * @throws IOException
     * @throws SerDeException
     */
    public List<orc.MyRow> readerOrc(String path,JobConf conf) throws IOException, SerDeException {
        StringBuilder str = new StringBuilder();
        List<orc.MyRow> stringList = new ArrayList<>();
        Path testFilePath = new Path(path);
        Properties p = new Properties();
        OrcSerde serde = new OrcSerde();
        p.setProperty("columns", "rowkey,family,field,value");
        p.setProperty("columns.types", "string:string:string:string");
        serde.initialize(conf, p);
        StructObjectInspector inspector = (StructObjectInspector) serde.getObjectInspector();
        InputFormat in = new OrcInputFormat();
        FileInputFormat.setInputPaths(conf, testFilePath.toString());
        InputSplit[] splits = in.getSplits(conf, 1);
        System.out.println("splits.length==" + splits.length);

        conf.set("hive.io.file.readcolumn.ids", "1");
        org.apache.hadoop.mapred.RecordReader reader = in.getRecordReader(splits[0], conf, Reporter.NULL);
        Object key = reader.createKey();
        Object value = reader.createValue();
        List<? extends StructField> fields = inspector.getAllStructFieldRefs();
        long offset = reader.getPos();
        while (reader.next(key, value)) {
            String rowkey = inspector.getStructFieldData(value, fields.get(0)).toString();
            String family = inspector.getStructFieldData(value, fields.get(1)).toString();
            Object field =  inspector.getStructFieldData(value, fields.get(2));
            Object fieldValue = inspector.getStructFieldData(value, fields.get(3));
            offset = reader.getPos();
            orc.MyRow myRow = new MyRow(rowkey, family,field.toString(),fieldValue.toString());
            stringList.add(myRow);
            str.append(rowkey).append("&").append(family).append("&").append(field).append("&").append(fieldValue).append("\t");
        }
        reader.close();
        return stringList;
    }

    /**
     * 导出orc文件
     *
     * @param path
     * @param myrow
     * @throws IOException
     */
    public static void writerOrc(Path path, MyRow myrow) throws IOException {
        JobConf conf = new JobConf();
        FileSystem fs = FileSystem.get(conf);
        // Path outputPath = new Path(path);
        StructObjectInspector inspector =
                (StructObjectInspector) ObjectInspectorFactory
                        .getReflectionObjectInspector(MyRow.class,
                                ObjectInspectorFactory.ObjectInspectorOptions.JAVA);
        OrcSerde serde = new OrcSerde();
        OutputFormat outFormat = new OrcOutputFormat();
        RecordWriter writer = outFormat.getRecordWriter(fs, conf,
                path.toString(), Reporter.NULL);
        writer.write(NullWritable.get(),
                serde.serialize(myrow, inspector));
        writer.close(Reporter.NULL);
        fs.close();
        System.out.println("write success .");
    }

    static class MyRow implements Writable {
        String rowkey;
        String family;
        String field;
        String value;

        public MyRow(String rowkey, String family, String field,String value) {
            this.rowkey = rowkey;
            this.family = family;
            this.field = field;
            this.value = value;
        }



        public String getValue() {
            return value;
        }

        public void setValue(String value) {
            this.value = value;
        }

        public String getRowkey() {
            return rowkey;
        }

        public void setRowkey(String rowkey) {
            this.rowkey = rowkey;
        }

        public String getFamily() {
            return family;
        }

        public void setFamily(String family) {
            this.family = family;
        }

        public String getField() {
            return field;
        }

        public void setField(String field) {
            this.field = field;
        }

        @Override
        public void write(DataOutput dataOutput) throws IOException {
            throw new UnsupportedOperationException("no read");
        }

        @Override
        public void readFields(DataInput dataInput) throws IOException {
            throw new UnsupportedOperationException("no write");
        }
    }

}
